﻿import importlib.util
import os
import sys
import math
import itertools
from typing import Callable, Dict, List, Tuple, Any, Iterable
from multiprocessing import cpu_count
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as _mp
from loguru import logger


class BacktestingService:
    """
    Service for running CTA strategy backtesting and parameter optimization.
    """

    def __init__(self) -> None:
        # Cache of last backtest records for UI retrieval
        self._last_trades: List[dict] = []
        self._last_orders: List[dict] = []
        self._last_daily_results: List[dict] = []

    def _load_strategy_and_engine(self, backtest_params: dict, log_callback: Callable[[str], None] = None):
        """Load strategy class and create configured BacktestingEngine."""
        from vnpy_ctastrategy.backtesting import BacktestingEngine

        strategy_dir = os.path.dirname(os.path.abspath(backtest_params["strategy_file_path"]))
        module_name = os.path.splitext(os.path.basename(backtest_params.get("strategy_file_path", "")))[0]

        # Add strategy directory to sys.path for dynamic import
        if strategy_dir not in sys.path:
            sys.path.insert(0, strategy_dir)
            logger.info(f"Added strategy directory to sys.path: {strategy_dir}")

        # Dynamically import the strategy class from the file
        spec = importlib.util.spec_from_file_location(
            module_name,
            backtest_params["strategy_file_path"]
        )
        if spec is None:
            raise ImportError(f"Could not load spec for module at {backtest_params['strategy_file_path']}")

        strategy_module = importlib.util.module_from_spec(spec)
        if spec.loader is None:
            raise ImportError(f"Could not get loader from spec for module at {backtest_params['strategy_file_path']}")

        # Pre-register module to sys.modules (helps multiprocessing pickling paths)
        sys.modules[module_name] = strategy_module
        spec.loader.exec_module(strategy_module)
        StrategyClass = getattr(strategy_module, backtest_params["strategy_name"])

        # Create and configure engine
        engine = BacktestingEngine()
        if log_callback is not None:
            engine.output = lambda msg: log_callback(str(msg))
        else:
            engine.output = lambda msg: logger.info(msg)

        engine.set_parameters(
            vt_symbol=backtest_params["vt_symbol"],
            interval=backtest_params["interval"],
            start=backtest_params["start_date"],
            end=backtest_params["end_date"],
            rate=backtest_params["rate"],
            slippage=backtest_params["slippage"],
            size=backtest_params["size"],
            pricetick=backtest_params["pricetick"],
            capital=backtest_params["capital"],
        )

        # Extract strategy-specific settings from backtest_params
        setting: Dict[str, Any] = {}
        reserved = {
            "strategy_name", "strategy_file_path", "vt_symbol", "interval",
            "start_date", "end_date", "rate", "slippage", "size", "pricetick", "capital"
        }
        for key, value in backtest_params.items():
            if key not in reserved:
                setting[key] = value

        engine.add_strategy(StrategyClass, setting)
        return StrategyClass, engine

    def _collect_backtest_data(self, engine) -> Tuple[Any, List[dict], List[dict], List[dict]]:
        """Collect backtest results data from the engine."""
        def _enum_to_str(e: Any) -> str:
            try:
                if hasattr(e, 'value'):
                    return str(e.value)
                return str(e)
            except Exception:
                return str(e)

        def _dt_to_str(dt: Any) -> str:
            try:
                if hasattr(dt, 'strftime'):
                    if hasattr(dt, 'hour'):
                        return dt.strftime('%Y-%m-%d %H:%M:%S')
                    return dt.strftime('%Y-%m-%d')
                return str(dt)
            except Exception:
                return str(dt)

        # Fetch history data for plotting
        history_data = None
        try:
            if hasattr(engine, 'history_data') and engine.history_data:
                history_data = engine.history_data
            elif hasattr(engine, 'bars') and engine.bars:
                history_data = engine.bars
        except Exception as e:
            logger.warning(f"Failed to get history data: {e}")

        # Collect trades
        trades_list = []
        try:
            trades_obj = None
            if hasattr(engine, 'get_all_trades'):
                trades_obj = engine.get_all_trades()  # type: ignore[attr-defined]
            elif hasattr(engine, 'trades'):
                trades_obj = engine.trades
            if trades_obj:
                iterable = trades_obj.values() if isinstance(trades_obj, dict) else trades_obj
                for t in iterable:
                    try:
                        trades_list.append({
                            'tradeid': t.tradeid,
                            'orderid': t.orderid,
                            'symbol': t.symbol,
                            'exchange': _enum_to_str(t.exchange),
                            'direction': _enum_to_str(t.direction),
                            'offset': _enum_to_str(t.offset),
                            'price': t.price,
                            'volume': t.volume,
                            'datetime': _dt_to_str(t.datetime),
                            'gateway_name': t.gateway_name,
                        })
                    except Exception:
                        continue
        except Exception as e:
            logger.exception(e)

        # Collect orders
        orders_list = []
        try:
            orders_obj = None
            if hasattr(engine, 'get_all_orders'):
                orders_obj = engine.get_all_orders()  # type: ignore[attr-defined]
            elif hasattr(engine, 'orders'):
                orders_obj = engine.orders
            if orders_obj:
                iterable = orders_obj.values() if isinstance(orders_obj, dict) else orders_obj
                for o in iterable:
                    try:
                        orders_list.append({
                            'orderid': o.orderid,
                            'symbol': o.symbol,
                            'exchange': _enum_to_str(o.exchange),
                            'type': _enum_to_str(o.type),
                            'direction': _enum_to_str(o.direction),
                            'offset': _enum_to_str(o.offset),
                            'price': o.price,
                            'volume': o.volume,
                            'traded': o.traded,
                            'status': _enum_to_str(o.status),
                            'datetime': _dt_to_str(o.datetime),
                            'gateway_name': o.gateway_name,
                        })
                    except Exception:
                        continue
        except Exception as e:
            logger.exception(e)

        # Collect daily results
        daily_results_list = []
        try:
            d_obj = None
            if hasattr(engine, 'get_all_daily_results'):
                d_obj = engine.get_all_daily_results()  # type: ignore[attr-defined]
            elif hasattr(engine, 'daily_results'):
                d_obj = engine.daily_results
            if d_obj:
                iterable = d_obj.values() if isinstance(d_obj, dict) else d_obj
                for dr in iterable:
                    try:
                        daily_results_list.append({
                            'date': _dt_to_str(dr.date),
                            'trade_count': dr.trade_count,
                            'start_pos': dr.start_pos,
                            'end_pos': dr.end_pos,
                            'turnover': dr.turnover,
                            'commission': dr.commission,
                            'slippage': dr.slippage,
                            'trading_pnl': dr.trading_pnl,
                            'holding_pnl': dr.holding_pnl,
                            'total_pnl': dr.total_pnl,
                            'net_pnl': dr.net_pnl,
                        })
                    except Exception:
                        continue
        except Exception as e:
            logger.exception(e)

        return history_data, trades_list, orders_list, daily_results_list

    def _validate_backtest_results(self, engine) -> Tuple[bool, str]:
        """Validate that backtest produced meaningful results."""
        # Check trade records exist
        try:
            trades_obj_check = None
            if hasattr(engine, 'get_all_trades'):
                trades_obj_check = engine.get_all_trades()  # type: ignore[attr-defined]
            elif hasattr(engine, 'trades'):
                trades_obj_check = engine.trades
            if not trades_obj_check:
                logger.warning("No trades generated during backtesting")
                return False, "No trades during backtest; check strategy or data range."
        except Exception:
            logger.warning("Failed to check trades; continue to calculation attempt")
        
        return True, ""

    def _calculate_backtest_results(self, engine) -> Tuple[bool, str, Any, Dict]:
        """Calculate and validate backtest results and statistics."""
        try:
            results = engine.calculate_result()
            statistics = engine.calculate_statistics()

            if results is None or getattr(results, 'empty', False):
                logger.warning("Empty results from backtesting engine")
                return False, "Empty backtest results; possibly no data or no trades.", None, None

            if statistics is None or not statistics:
                logger.warning("Empty statistics from backtesting engine")
                return False, "Empty statistics; possibly no data or no trades.", None, None

            return True, "", results, statistics

        except KeyError as ke:
            if "date" in str(ke):
                logger.exception(ke)
                return False, (
                    "Backtest result empty: 1) no trade data in range 2) strategy made no trades "
                    "3) wrong symbol"
                ), None, None
            else:
                raise ke
        except Exception as calc_error:
            logger.exception(calc_error)
            return False, f"Error calculating backtest results: {calc_error}", None, None

    def _cleanup_strategy_imports(self, backtest_params: dict) -> None:
        """Clean up strategy-related imports and sys.path modifications."""
        module_name = os.path.splitext(os.path.basename(backtest_params.get("strategy_file_path", "")))[0]
        strategy_dir = os.path.dirname(os.path.abspath(backtest_params["strategy_file_path"]))

        # Remove strategy directory from sys.path
        try:
            if strategy_dir in sys.path:
                sys.path.remove(strategy_dir)
                logger.info(f"Removed strategy directory from sys.path: {strategy_dir}")
        except Exception as cleanup_error:
            logger.exception(cleanup_error)

        # Remove strategy module from sys.modules
        if module_name and module_name in sys.modules:
            try:
                del sys.modules[module_name]
                logger.info(f"Removed strategy module from sys.modules: {module_name}")
            except Exception as cleanup_error:
                logger.exception(cleanup_error)

    def _cleanup_database_connections(self) -> None:
        """Clean up previous database connections to avoid conflicts."""
        try:
            # Close existing peewee database connections
            try:
                if 'vnpy.trader.database' in sys.modules:
                    try:
                        db_obj = getattr(sys.modules['vnpy.trader.database'], 'database', None)
                        if db_obj is not None:
                            peewee_db = getattr(db_obj, 'db', None)
                            if peewee_db is not None:
                                try:
                                    peewee_db.close()
                                except Exception as close_err:
                                    logger.exception(close_err)
                    except Exception as inner_err:
                        logger.exception(inner_err)

                if 'vnpy_sqlite.sqlite_database' in sys.modules:
                    try:
                        sqlite_mod = sys.modules['vnpy_sqlite.sqlite_database']
                        peewee_db2 = getattr(sqlite_mod, 'db', None)
                        if peewee_db2 is not None:
                            try:
                                peewee_db2.close()
                            except Exception as close_err:
                                logger.exception(close_err)
                    except Exception as inner_err:
                        logger.exception(inner_err)
            except Exception as close_outer_err:
                logger.exception(close_outer_err)

            # Clean up module references
            if 'vnpy.trader.database' in sys.modules:
                try:
                    sys.modules['vnpy.trader.database'].database = None  # type: ignore
                except Exception:
                    pass
                del sys.modules['vnpy.trader.database']
            for m in ['vnpy_sqlite.sqlite_database', 'vnpy.trader.utility']:
                if m in sys.modules:
                    del sys.modules[m]
        except Exception as e:
            logger.exception(e)

    def run_backtest(self, backtest_params: dict, workspace_path: str, log_callback: Callable[[str], None] = None):
        """
        Run a single backtest with the given parameters.
        Returns tuple:
          (success: bool,
           message: str,
           results: Any,            # equity curve DataFrame
           statistics: dict,        # summary stats
           history_data: Any,       # historical bars (list/df)
           trades_list: list[dict], # 成交记录
           orders_list: list[dict], # 委托记录
           daily_results: list[dict]) # 每日盈亏
        """
        original_cwd = os.getcwd()
        os.chdir(workspace_path)

        # Ensure vn.py binds to <workspace>/.vntrader
        self._cleanup_database_connections()

        try:
            strategy_class, engine = self._load_strategy_and_engine(backtest_params, log_callback)

            # Run backtest
            engine.load_data()
            engine.run_backtesting()

            # Validate results
            is_valid, error_msg = self._validate_backtest_results(engine)
            if not is_valid:
                return False, error_msg, None, None, None, None, None, None

            # Calculate results and statistics
            success, error_msg, results, statistics = self._calculate_backtest_results(engine)
            if not success:
                return False, error_msg, None, None, None, None, None, None

            # Collect data
            history_data, trades_list, orders_list, daily_results_list = self._collect_backtest_data(engine)

            # Cache for later retrieval via getters
            try:
                self._last_trades = list(trades_list)
                self._last_orders = list(orders_list)
                self._last_daily_results = list(daily_results_list)
            except Exception:
                pass

            logger.info("Backtesting finished successfully.")
            return True, "Backtest completed", results, statistics, history_data, trades_list, orders_list, daily_results_list

        except ImportError as ie:
            logger.exception(ie)
            return False, f"Failed to import strategy file: {ie}", None, None, None, None, None, None
        except AttributeError as ae:
            logger.exception(ae)
            return False, f"Attribute error: {ae}", None, None, None, None, None, None
        except Exception as e:
            logger.exception(e)
            return False, f"Unexpected error during backtest: {e}", None, None, None, None, None, None
        finally:
            # Cleanup strategy-related imports and sys.path
            self._cleanup_strategy_imports(backtest_params)
            os.chdir(original_cwd)

    # -------- Record Getters (参考 widget.py 的方式) --------
    def get_all_trades(self) -> List[dict]:
        """获取最近一次回测的成交记录。"""
        return list(self._last_trades)

    def get_all_orders(self) -> List[dict]:
        """获取最近一次回测的委托记录。"""
        return list(self._last_orders)

    def get_all_daily_results(self) -> List[dict]:
        """获取最近一次回测的每日盈亏汇总。"""
        return list(self._last_daily_results)

    def run_optimization(self, backtest_params: dict, optimization_setting: Any,
                         workspace_path: str, use_ga: bool = False, max_workers: int = 0,
                         progress_callback: Callable[[int, int, str], None] = None) -> List[Tuple[Dict, float, Dict]]:
        """
        Run parameter optimization. For large grids, use streaming brutal-force optimization
        to avoid building the full Cartesian product in memory.

        Returns [(setting, target_value, statistics), ...]
        """
        original_cwd = os.getcwd()
        os.chdir(workspace_path)

        # Reset vn.py modules to bind to <workspace>/.vntrader
        try:
            # 先尝试关闭残留数据库连接，避免二次 connect 抛错
            try:
                if 'vnpy.trader.database' in sys.modules:
                    try:
                        db_obj = getattr(sys.modules['vnpy.trader.database'], 'database', None)
                        if db_obj is not None:
                            peewee_db = getattr(db_obj, 'db', None)
                            if peewee_db is not None:
                                try:
                                    peewee_db.close()
                                except Exception as close_err:
                                    logger.exception(close_err)
                    except Exception as inner_err:
                        logger.exception(inner_err)

                if 'vnpy_sqlite.sqlite_database' in sys.modules:
                    try:
                        sqlite_mod = sys.modules['vnpy_sqlite.sqlite_database']
                        peewee_db2 = getattr(sqlite_mod, 'db', None)
                        if peewee_db2 is not None:
                            try:
                                peewee_db2.close()
                            except Exception as close_err:
                                logger.exception(close_err)
                    except Exception as inner_err:
                        logger.exception(inner_err)
            except Exception as close_outer_err:
                logger.exception(close_outer_err)

            if 'vnpy.trader.database' in sys.modules:
                try:
                    sys.modules['vnpy.trader.database'].database = None  # type: ignore
                except Exception:
                    pass
                del sys.modules['vnpy.trader.database']
            for m in ['vnpy_sqlite.sqlite_database', 'vnpy.trader.utility']:
                if m in sys.modules:
                    del sys.modules[m]
        except Exception as e:
            logger.exception(e)

        strategy_dir = None
        added_strategy_dir_to_syspath = False
        module_name = os.path.splitext(os.path.basename(backtest_params.get("strategy_file_path", "")))[0]

        # Prefer streaming path for brutal-force search
        if not use_ga:
            try:
                results = self._run_optimization_streaming(
                    backtest_params,
                    optimization_setting,
                    max_workers,
                    progress_callback,
                )
                os.chdir(original_cwd)
                return results
            except Exception as _stream_err:
                logger.warning(f"Streaming optimization failed, fallback to vn.py optimize: {_stream_err}")

        try:
            from vnpy_ctastrategy.backtesting import BacktestingEngine
            from vnpy.trader.optimize import OptimizationSetting as VnOptimizationSetting

            # Add strategy dir to sys.path
            strategy_dir = os.path.dirname(os.path.abspath(backtest_params["strategy_file_path"]))
            if strategy_dir not in sys.path:
                sys.path.insert(0, strategy_dir)
                added_strategy_dir_to_syspath = True

            # Load strategy class dynamically
            spec = importlib.util.spec_from_file_location(
                module_name,
                backtest_params["strategy_file_path"]
            )
            if spec is None or spec.loader is None:
                raise ImportError(f"Could not load strategy module at {backtest_params['strategy_file_path']}")
            strategy_module = importlib.util.module_from_spec(spec)
            sys.modules[module_name] = strategy_module
            spec.loader.exec_module(strategy_module)
            StrategyClass = getattr(strategy_module, backtest_params["strategy_name"])

            engine = BacktestingEngine()
            # Redirect engine logs to progress callback if provided
            if progress_callback is not None:
                def _output(msg: str) -> None:
                    try:
                        progress_callback(0, 0, str(msg))
                    except Exception as cb_err:
                        logger.exception(cb_err)
                engine.output = _output
            else:
                engine.output = lambda msg: logger.info(msg)

            # Base backtest parameters
            engine.set_parameters(
                vt_symbol=backtest_params["vt_symbol"],
                interval=backtest_params["interval"],
                start=backtest_params["start_date"],
                end=backtest_params["end_date"],
                rate=backtest_params["rate"],
                slippage=backtest_params["slippage"],
                size=backtest_params["size"],
                pricetick=backtest_params["pricetick"],
                capital=backtest_params["capital"],
            )

            # Strategy defaults from backtest_params
            strategy_setting: Dict[str, Any] = {}
            for key, value in backtest_params.items():
                if key not in [
                    "strategy_name", "strategy_file_path", "vt_symbol", "interval",
                    "start_date", "end_date", "rate", "slippage", "size", "pricetick", "capital"
                ]:
                    strategy_setting[key] = value
            engine.add_strategy(StrategyClass, strategy_setting)

            # Ensure child processes spawned by vn.py can import sibling modules of the strategy file
            try:
                parent_dir = os.path.dirname(strategy_dir) if strategy_dir else None
                cur_env = os.environ.get("PYTHONPATH", "")
                need_add: List[str] = []
                for p in [strategy_dir, parent_dir]:
                    if p and p not in cur_env.split(os.pathsep):
                        need_add.append(p)
                if need_add:
                    os.environ["PYTHONPATH"] = cur_env + ((os.pathsep if cur_env else "") + os.pathsep.join(need_add))
            except Exception as e:
                logger.exception(e)

            # Convert our OptimizationSetting or dict to vn.py's
            if hasattr(optimization_setting, 'to_dict'):
                opt_dict: Dict[str, Any] = optimization_setting.to_dict()  # type: ignore
            elif isinstance(optimization_setting, dict):
                opt_dict = optimization_setting
            else:
                opt_dict = {}

            vn_opt = VnOptimizationSetting()
            target_name: str = opt_dict.get('target', 'total_return')
            vn_opt.set_target(target_name)

            parameters: Dict[str, Any] = opt_dict.get('parameters', {})
            for name, cfg in parameters.items():
                try:
                    if isinstance(cfg, dict) and 'values' in cfg:
                        vn_opt.add_parameter(name, cfg.get('values') or [])
                        continue

                    start = cfg.get('start') if isinstance(cfg, dict) else None
                    end = cfg.get('end') if isinstance(cfg, dict) else None
                    step = cfg.get('step') if isinstance(cfg, dict) else None

                    if start is None and end is None and step is None:
                        logger.warning(f"Parameter '{name}' has no valid range or values; skipped")
                        continue

                    if end is None or step is None or start == end:
                        vn_opt.add_parameter(name, start)
                    else:
                        vn_opt.add_parameter(name, start, end, step)
                except Exception as e:
                    logger.exception(e)

            workers = None if max_workers in (0, None) else max_workers

            # Call vn.py optimization
            if use_ga:
                # GA requires at least 2 parameters
                if len(getattr(vn_opt, 'params', [])) < 2:
                    logger.warning(
                        f"GA needs at least 2 parameters; only {len(getattr(vn_opt, 'params', []))} provided. "
                        f"Fallback to brute-force optimization."
                    )
                    results = engine.run_optimization(vn_opt, output=True, max_workers=workers)
                else:
                    results = engine.run_ga_optimization(vn_opt, output=True, max_workers=workers)
            else:
                results = engine.run_optimization(vn_opt, output=True, max_workers=workers)

            return results

        except Exception as e:
            logger.exception(e)
            raise
        finally:
            if added_strategy_dir_to_syspath and strategy_dir:
                try:
                    sys.path.remove(strategy_dir)
                except Exception as cleanup_error:
                    logger.exception(cleanup_error)
            if module_name and module_name in sys.modules:
                try:
                    del sys.modules[module_name]
                except Exception as cleanup_error:
                    logger.exception(cleanup_error)
            os.chdir(original_cwd)

    def _run_optimization_streaming(
        self,
        backtest_params: dict,
        optimization_setting: Any,
        max_workers: int,
        progress_callback: Callable[[int, int, str], None] | None,
    ) -> List[Tuple[Dict, float, Dict]]:
        """
        Streaming (no full expansion) brutal-force optimization to avoid large grids causing crashes.
        """
        # Normalize to dict
        if hasattr(optimization_setting, 'to_dict'):
            opt_dict: Dict[str, Any] = optimization_setting.to_dict()  # type: ignore
        elif isinstance(optimization_setting, dict):
            opt_dict = optimization_setting
        else:
            opt_dict = {}

        target_name: str = opt_dict.get('target', 'total_return')
        parameters_cfg: Dict[str, Any] = opt_dict.get('parameters', {})

        # Build value lists for each parameter
        param_names: List[str] = []
        value_lists: List[List[Any]] = []
        for name, cfg in parameters_cfg.items():
            values: List[Any] = []
            if isinstance(cfg, dict) and 'values' in cfg and isinstance(cfg['values'], (list, tuple)):
                values = list(cfg['values'])
            else:
                if not isinstance(cfg, dict):
                    continue
                start = cfg.get('start')
                end = cfg.get('end')
                step = cfg.get('step')
                if step in (None, 0) or end is None or start == end:
                    values = [start]
                else:
                    try:
                        count = int(math.floor((end - start) / step)) + 1
                        if count <= 0:
                            values = [start]
                        else:
                            values = [start + i * step for i in range(count)]
                            if abs(values[-1] - end) <= (abs(step) * 1e-9 + 1e-12):
                                values[-1] = end
                    except Exception:
                        values = [start]
            if not values:
                continue
            param_names.append(name)
            value_lists.append(values)

        # Total combinations
        total: int = 1
        for seq in value_lists:
            total *= max(1, len(seq))

        if progress_callback is not None:
            try:
                progress_callback(0, total, f"Start optimization: total combinations {total}")
            except Exception:
                pass

        # Combination generator
        def combos_iter() -> Iterable[Dict[str, Any]]:
            for tup in itertools.product(*value_lists) if value_lists else [()]:
                yield dict(zip(param_names, tup))

        # Base args passed to workers (reduce pickling overhead)
        base_args = {
            "target_name": target_name,
            "strategy_file_path": os.path.abspath(backtest_params["strategy_file_path"]),
            "strategy_name": backtest_params["strategy_name"],
            "vt_symbol": backtest_params["vt_symbol"],
            "interval": backtest_params["interval"],
            "start": backtest_params["start_date"],
            "end": backtest_params["end_date"],
            "rate": backtest_params["rate"],
            "slippage": backtest_params["slippage"],
            "size": backtest_params["size"],
            "pricetick": backtest_params["pricetick"],
            "capital": backtest_params["capital"],
        }

        # Ensure children can import strategy file directory and its parent (for sibling utilities)
        strategy_dir = os.path.dirname(os.path.abspath(backtest_params["strategy_file_path"]))
        parent_dir = os.path.dirname(strategy_dir) if strategy_dir else None
        if strategy_dir:
            cur = os.environ.get("PYTHONPATH", "")
            add_paths: List[str] = []
            for p in [strategy_dir, parent_dir]:
                if p and p not in cur.split(os.pathsep):
                    add_paths.append(p)
            if add_paths:
                os.environ["PYTHONPATH"] = cur + ((os.pathsep if cur else "") + os.pathsep.join(add_paths))

        ctx = _mp.get_context("spawn")
        workers = (cpu_count() - 1) if max_workers in (0, None) else max(1, int(max_workers))
        chunksize = 1
        if total and workers:
            chunksize = max(1, min(256, total // (workers * 8) or 1))

        def tasks_gen() -> Iterable[Tuple[Dict[str, Any], Dict[str, Any]]]:
            for s in combos_iter():
                yield (base_args, s)

        results: List[Tuple[Dict, float, Dict]] = []
        processed = 0
        next_progress_emit = 0
        with ProcessPoolExecutor(max_workers=workers, mp_context=ctx) as executor:
            for res in executor.map(_evaluate_optimization_worker, tasks_gen(), chunksize=chunksize):
                processed += 1
                if isinstance(res, tuple) and len(res) == 3:
                    results.append(res)
                if progress_callback is not None and total:
                    threshold = max(100, total // 100)
                    if processed >= next_progress_emit:
                        try:
                            progress_callback(processed, total, f"Progress {processed}/{total}")
                        except Exception:
                            pass
                        next_progress_emit = processed + threshold

        try:
            results.sort(key=lambda x: x[1], reverse=True)
        except Exception:
            pass

        if progress_callback is not None:
            try:
                progress_callback(processed, total, f"Done, results {len(results)}")
            except Exception:
                pass

        return results

    def _generate_parameter_combinations(self, parameters: Dict[str, Dict]) -> List[Dict]:
        """Deprecated: kept for compatibility."""
        return []

    def _run_parallel_optimization(self, backtest_params: dict, target_name: str,
                                   parameters: Dict[str, Dict], workspace_path: str,
                                   max_workers: int, progress_callback: Callable) -> List[Tuple[Dict, float, Dict]]:
        """Deprecated: replaced by streaming optimization."""
        return []

    def _run_ga_optimization(self, backtest_params: dict, optimization_setting: Dict[str, Any],
                              workspace_path: str, max_workers: int,
                              progress_callback: Callable) -> List[Tuple[Dict, float, Dict]]:
        """Deprecated: use vn.py's GA optimization via run_optimization."""
        return []

    def get_strategy_default_parameters(self, strategy_file_path: str, strategy_name: str) -> Dict[str, Any]:
        """
        Read default parameter values from strategy class.
        Supports vn.py convention: class attribute 'parameters' lists names, with class attributes as defaults.
        Falls back to __init__ defaults if class attributes not present.
        """
        try:
            module_name = os.path.splitext(os.path.basename(strategy_file_path))[0]
            strategy_dir = os.path.dirname(os.path.abspath(strategy_file_path))
            parent_dir = os.path.dirname(strategy_dir) if strategy_dir else None

            added_paths: List[str] = []
            # Ensure imports like `from daily_generator import ...` work for sibling files
            for path in [strategy_dir, parent_dir]:
                if path and path not in sys.path:
                    sys.path.insert(0, path)
                    added_paths.append(path)

            spec = importlib.util.spec_from_file_location(module_name, strategy_file_path)
            if spec is None or spec.loader is None:
                # Cleanup inserted paths before returning
                for p in added_paths:
                    try:
                        sys.path.remove(p)
                    except Exception:
                        pass
                return {}

            strategy_module = importlib.util.module_from_spec(spec)
            sys.modules[module_name] = strategy_module
            spec.loader.exec_module(strategy_module)
            StrategyClass = getattr(strategy_module, strategy_name)

            import inspect
            ctor_sig = inspect.signature(StrategyClass.__init__)
            ctor_defaults: Dict[str, Any] = {}
            for param_name, param in ctor_sig.parameters.items():
                if param_name in ['self', 'cta_engine', 'strategy_name', 'vt_symbol', 'setting']:
                    continue
                if param.default != inspect.Parameter.empty:
                    ctor_defaults[param_name] = param.default

            if hasattr(StrategyClass, 'parameters'):
                params_attr = getattr(StrategyClass, 'parameters')
                if isinstance(params_attr, dict):
                    return params_attr

                if isinstance(params_attr, (list, tuple)):
                    result: Dict[str, Any] = {}
                    for name in params_attr:
                        if hasattr(StrategyClass, name):
                            result[name] = getattr(StrategyClass, name)
                        elif name in ctor_defaults:
                            result[name] = ctor_defaults[name]
                        else:
                            result[name] = 0
                    return result

            return ctor_defaults

        except Exception as e:
            logger.exception(e)
            return {}
        finally:
            # Remove any sys.path entries we added
            try:
                if 'added_paths' in locals():
                    for p in added_paths:
                        try:
                            sys.path.remove(p)
                        except Exception:
                            pass
            except Exception:
                pass


def _evaluate_optimization_worker(args: Tuple[Dict[str, Any], Dict[str, Any]]) -> Tuple[Dict, float, Dict]:
    """
    Worker: run a single backtest for a given parameter combination.
    Returns (setting, target_value, statistics).
    """
    try:
        base_args, setting = args

        from vnpy_ctastrategy.backtesting import BacktestingEngine

        # 确保工作进程可以导入策略同目录/父目录中的模块（如 daily_generator）
        try:
            strategy_dir = os.path.dirname(os.path.abspath(base_args["strategy_file_path"]))
            parent_dir = os.path.dirname(strategy_dir) if strategy_dir else None
            for p in [strategy_dir, parent_dir]:
                if p and p not in sys.path:
                    sys.path.insert(0, p)
        except Exception as add_path_err:
            logger.exception(add_path_err)

        module_name = os.path.splitext(os.path.basename(base_args["strategy_file_path"]))[0]
        spec = importlib.util.spec_from_file_location(module_name, base_args["strategy_file_path"])
        if spec is None or spec.loader is None:
            raise ImportError(f"Could not load strategy module: {base_args['strategy_file_path']}")
        strategy_module = importlib.util.module_from_spec(spec)
        sys.modules[module_name] = strategy_module
        spec.loader.exec_module(strategy_module)

        StrategyClass = getattr(strategy_module, base_args["strategy_name"])

        engine = BacktestingEngine()
        engine.output = lambda msg: None

        engine.set_parameters(
            vt_symbol=base_args["vt_symbol"],
            interval=base_args["interval"],
            start=base_args["start"],
            end=base_args["end"],
            rate=base_args["rate"],
            slippage=base_args["slippage"],
            size=base_args["size"],
            pricetick=base_args["pricetick"],
            capital=base_args["capital"],
        )

        engine.add_strategy(StrategyClass, setting)
        engine.load_data()
        engine.run_backtesting()
        engine.calculate_result()
        statistics: Dict[str, Any] = engine.calculate_statistics(output=False)

        target_value = statistics.get(base_args["target_name"]) if isinstance(statistics, dict) else None
        if target_value is None:
            target_value = float("-inf")

        return (setting, float(target_value), statistics)
    except Exception as e:
        return ({"__error__": str(e)}, float("-inf"), {"error": str(e)})


# Compatibility helper for UI dialog
class OptimizationSetting:
    """Simple optimization setting container compatible with UI."""

    def __init__(self):
        self.target: str = "total_return"
        self.parameters: Dict[str, Dict] = {}

    def set_target(self, target_name: str):
        self.target = target_name

    def add_parameter(self, name: str, start=None, end=None, step=None, values=None):
        if values is not None:
            self.parameters[name] = {"values": values}
        else:
            self.parameters[name] = {
                "start": start,
                "end": end,
                "step": step
            }

    def to_dict(self) -> Dict[str, Any]:
        return {
            "target": self.target,
            "parameters": self.parameters
        }

